package org.anno.chunkdb.backend;

import org.anno.chunkdb.Chunkdb;

import java.util.Arrays;
import java.util.Comparator;
import java.util.List;
import java.util.function.IntFunction;
import java.util.function.IntPredicate;
import java.util.function.Predicate;
import java.util.function.ToIntFunction;
import java.util.stream.IntStream;
import java.util.stream.Stream;

/**
 * Common interface for streaming over rows, which have a value of a specific type for each column.
 */
public interface Table<R> {

  List<Column<R>> getColumns();

  Stream<R> getAll();

  /**
   * Selects rows sorted by values in the indicated column in the half open interval, min <= v <
   * max.
   */
  default Stream<R> getSorted(Table<R> table, int column, double min, double max, boolean asc) {
    return table.getColumns().get(column).getRowRange(min, max, asc).mapToObj(
        getIntToRowFunction());
  }

  /**
   * Selects rows sorted by values in the indicated column in the half open interval, min <= v <
   * max.
   */
  default Stream<R> getSorted(Table<R> table, int column, byte[] min, byte[] max, boolean asc) {
    return table.getColumns().get(column).getRowRange(min, max, asc).mapToObj(
        getIntToRowFunction());
  }

  IntFunction<R> getIntToRowFunction();

  interface Column<R> {

    int rankToOffset(int i);

    Comparator<R> getComparator();

    default Predicate<R> getPredicate(Chunkdb.Filter filter) {
      IntPredicate p = getDiffPredicate(filter.getOp());
      ToIntFunction<R> diff = getDiffFunction(filter);
      return r->p.test(diff.applyAsInt(r));
    }

    default IntPredicate getDiffPredicate(Chunkdb.Filter.Op op) {
      switch (op) {
        case LT:
          return i -> i < 0;
        case LE:
          return i -> i <= 0;
        case EQ:
          return i -> i == 0;
        case GE:
          return i -> i >= 0;
        case GT:
          return i -> i > 0;
      }
      throw new IllegalArgumentException("op=" + op);
    }

    ToIntFunction<R> getDiffFunction(Chunkdb.Filter filter);

    default double getDouble(R row) {
      throw new IllegalArgumentException("No doubles here");
    }

    default ByteSlice getString(R row) {
      throw new IllegalArgumentException("No strings here");
    }

    default int getGERank(double min) {
      throw new IllegalArgumentException("No doubles here");
    }

    default int getGTRank(double min) {
      throw new IllegalArgumentException("No doubles here");
    }

    default int getLERank(double max) {
      throw new IllegalArgumentException("No doubles here");
    }

    default int getLTRank(double max) {
      throw new IllegalArgumentException("No doubles here");
    }

    default int getGERank(byte[] min) {
      throw new IllegalArgumentException("No strings here");
    }

    default int getGTRank(byte[] min) {
      return getGERank(Arrays.copyOf(min, min.length + 1));
    }

    default int getLERank(byte[] max) {
      return getLTRank(Arrays.copyOf(max, max.length + 1));
    }

    default int getLTRank(byte[] max) {
      throw new IllegalArgumentException("No strings here");
    }

    default IntStream getRowRange(byte[] min, byte[] max, boolean asc) {
      return getRowRange(getGERank(min), getLTRank(max) + 1, asc);
    }

    default IntStream getRowRange(double min, double max, boolean asc) {
      return getRowRange(getGERank(min), getLTRank(max) + 1, asc);
    }

    default IntStream getRowRange(int minIndex, int maxIndex, boolean asc) {
      IntStream range = IntStream.range(minIndex, maxIndex);
      if (!asc) {
        int last = maxIndex + minIndex - 1;
        range = range.map(i -> last - i);
      }
      return range.map(this::rankToOffset);
    }
  }
}
